package com.deng.rtc.mq;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import com.deng.rtc.base.MqConsumer;
import com.deng.rtc.util.Constants;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;
import kafka.serializer.StringDecoder;
import kafka.utils.VerifiableProperties;

public class KafkaMqCollect extends MqConsumer {
	// 定义消费者
	private ConsumerConnector consumer;
	
	public KafkaMqCollect(){
        // 父类初始化
		super();
		this.init();
	}
	
	public void init(){
		Properties props = new Properties();
        props.put("zookeeper.connect", Constants.kfZkServers);
        props.put("group.id", Constants.kfGroupId);
        props.put("auto.offset.reset", Constants.kfAutoOffsetReset);
        props.put("zookeeper.session.timeout.ms", "4000");
        props.put("zookeeper.sync.time.ms", "200");
        props.put("auto.commit.interval.ms", "1000");
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        ConsumerConfig config = new ConsumerConfig(props);
        consumer = Consumer.createJavaConsumerConnector(config);
	}

	// 实时收集数据
	public void collectMq(){
		Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put(Constants.kfTopic, new Integer(1));
 
        StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());
        StringDecoder valueDecoder = new StringDecoder(new VerifiableProperties());
 
        Map<String, List<KafkaStream<String, String>>> consumerMap =
                consumer.createMessageStreams(topicCountMap,keyDecoder,valueDecoder);
        
        KafkaStream<String, String> stream = consumerMap.get(Constants.kfTopic).get(0);
        ConsumerIterator<String, String> it = stream.iterator();
        MessageAndMetadata<String, String> msgMeta;
        while (it.hasNext()){
        	msgMeta = it.next();
            // 调用父类方法，同步相同主键数据定位到固定AN对应的本地队列
        	super.mqTimer.parseMqText(msgMeta.key(), msgMeta.message());
        	System.out.println(msgMeta.key()+"\t"+msgMeta.message());
        }
	}

//	public static void main(String[] args) {
//		KafkaMqCollect kmc = new KafkaMqCollect();
//		kmc.collectMq();
//	}

}
